package com.shujia.hbaseapi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

/**
 *  组合过滤器的使用
 *
 *
 *  1 通过运用4种比较器过滤出姓于，年纪大于23岁，性别为女，且是理科的学生。
 *
 *
 *
 *  2 过滤出学号是以15001001开头的文科学生
 */
public class HbaseFileCombinationDemo {
    private HConnection conn;
    private HBaseAdmin hAdmin;

    @Before
    public void connect() {
        try {
            //1、获取Hadoop的相关配置环境
            Configuration conf = new Configuration();

            //2、获取zookeeper的配置
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
            //获取与Hbase的连接，这个连接是将来可以用户获取hbase表的
            conn = HConnectionManager.createConnection(conf);

            //将来我们要对表做DDL相关操作，而对表的操作在hbase架构中是有HMaster
            hAdmin = new HBaseAdmin(conf);

            System.out.println("建立连接成功:" + conn + ", HMaster获取成功：" + hAdmin);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求：1 通过运用4种比较器过滤出姓于，年纪大于23岁，性别为女，且是理科的学生。
     *
     * 正则比较器   RegexStringComparator
     * 包含比较器   SubstringComparator
     * 二进制前缀比较器   BinaryPrefixComparator
     * 二进制比较器      BinaryComparator
     *
     */
    @Test
    public void FilterData1(){
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            /**
             *  第一个过滤器，过滤出是理科开头的班级
             */
            RegexStringComparator regexStringComparator = new RegexStringComparator("^理科.*");
            //单列值过滤器
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("info".getBytes(), "clazz".getBytes(),
                    CompareFilter.CompareOp.EQUAL, regexStringComparator);

            /**
             * 第二个过滤器，过滤出性别是女生的
             */

            SubstringComparator substringComparator = new SubstringComparator("女");
            SingleColumnValueFilter singleColumnValueFilter1 = new SingleColumnValueFilter("info".getBytes(), "gender".getBytes(),
                    CompareFilter.CompareOp.EQUAL, substringComparator);

            /**
             * 第三个过滤器，过滤出年龄大于23岁的
             */
            BinaryComparator binaryComparator = new BinaryComparator("20".getBytes());
            SingleColumnValueFilter singleColumnValueFilter2 = new SingleColumnValueFilter("info".getBytes(), "age".getBytes(),
                    CompareFilter.CompareOp.GREATER, binaryComparator);


            /**
             * 第四个过滤器，过滤出姓于的学生
             */
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("于".getBytes());
            SingleColumnValueFilter singleColumnValueFilter3 = new SingleColumnValueFilter("info".getBytes(), "name".getBytes(),
                    CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);


            Scan scan = new Scan();

            //要想实现多个需求同时过滤，就需要创建多个过滤器，添加到一个过滤器列表中
            //然后将过滤器列表传给扫描器scan
            FilterList filterList = new FilterList();
            filterList.addFilter(singleColumnValueFilter);
            filterList.addFilter(singleColumnValueFilter1);
            filterList.addFilter(singleColumnValueFilter2);
            filterList.addFilter(singleColumnValueFilter3);

            scan.setFilter(filterList);

            ResultScanner scanner = students.getScanner(scan);

            print2(scanner);


        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    /**
     * 过滤出学号是以15001001开头的文科学生
     */
    @Test
    public void filterData2(){
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            /**
             *  创建第一个过滤器，过滤是以15001001开头的rowkey
             */
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("15001001".getBytes());
            //创建行键过滤器
            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

            /**
             * 创建第二个过滤器，过滤出文科的学生
             */
            RegexStringComparator regexStringComparator = new RegexStringComparator("^文科.*");
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("info".getBytes(), "clazz".getBytes(),
                    CompareFilter.CompareOp.EQUAL,
                    regexStringComparator);


            FilterList filterList = new FilterList();
            filterList.addFilter(rowFilter);
            filterList.addFilter(singleColumnValueFilter);

            Scan scan = new Scan();
            scan.setFilter(filterList);
            ResultScanner scanner = students.getScanner(scan);
            print2(scanner);

        } catch (IOException e) {
            e.printStackTrace();
        }

    }




    /**
     * 专门用来打印数据的方法
     */
    public void print(ResultScanner scanner) throws IOException {
        Result rs = null;
        while ((rs = scanner.next()) != null) {
            String id = Bytes.toString(rs.getRow());
            List<Cell> cells = rs.listCells();
            System.out.print("id:" + id);
            System.out.print("\t");
            for (Cell cell : cells) {
//                String s = Bytes.toString(cell.getValue());
                String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                String s = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(col + ":" + s);
                System.out.print("\t");
            }
            System.out.println();

//            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
//            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
//            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
//            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
//            System.out.println("学号：" + id + ",姓名：" + name + ",年龄：" + age + ",性别：" + gender + ",班级：" + clazz);

        }
    }


    /**
     * 专门用来打印数据的方法
     */
    public void print2(ResultScanner scanner) throws IOException {
        Result rs = null;
        while ((rs = scanner.next()) != null) {
            String id = Bytes.toString(rs.getRow());

            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println("学号：" + id + ",姓名：" + name + ",年龄：" + age + ",性别：" + gender + ",班级：" + clazz);

        }
    }


    @After
    public void close() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("conn连接已经关闭.....");
        }

        if (hAdmin != null) {
            try {
                hAdmin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("HMaster已经关闭......");
        }
    }
}
